#include <brpc/server.h>
#include <butil/logging.h>
#include <map>
//实现语音识别子服务
#include <unordered_set>
#include "mysql_message.hpp"
#include "data_es.hpp"
#include "data_redis.hpp"
#include "utils.hpp"
#include "etcd.hpp"     //服务注册
#include "logger.hpp"   //日志模块
#include "channel.hpp"
#include "rabbitmq.hpp"

#include "message.pb.h"
#include "base.pb.h"
#include "file.pb.h"    //基于文件子服务进行上传下载
#include "user.pb.h"    //基于文件子服务进行上传下载

namespace xchat{
//proto rpc SpeechService ; package xchat
class MessageServiceImpl : public xchat::MsgStorageService
{
public:
    MessageServiceImpl(const std::shared_ptr<elasticlient::Client> &es_client
                    ,const std::shared_ptr<odb::core::database> &mysql_client
                    ,const ServiceManager::ptr &channel_manager
                    ,const std::string &file_service_name
                    ,const std::string &user_service_name)
                    :_es_message(std::make_shared<ESMessage>(es_client))
                    ,_mysql_message(std::make_shared<MessageTable>(mysql_client))
                    ,_file_service_name(file_service_name)
                    ,_user_service_name(user_service_name)
                    ,_mm_channels(channel_manager)

    {
                _es_message->createIndex();
    }
    ~MessageServiceImpl(){}
    //给出会话ID和起始时间 从数据库拿出消息的属性 用文件ID去文件子服务拿文件内容 用用户ID去用户子服务拿用户信息
    void GetHistoryMsg(::google::protobuf::RpcController* controller,
                       const ::xchat::GetHistoryMsgReq* request,
                       ::xchat::GetHistoryMsgRsp* response,
                       ::google::protobuf::Closure* done)
    {
        brpc::ClosureGuard rpc_guard(done);
        auto error_response = [this, response](const std::string &rid, const std::string &error_msg) -> void
        {
            response->set_request_id(rid);
            response->set_success(false);
            response->set_errmsg(error_msg);
        };
        // 1.提取关键要素 会话ID 起始时间 结束时间
        std::string rid = request->request_id();
        std::string chat_ssid = request->chat_session_id();
        boost::posix_time::ptime start_time = boost::posix_time::from_time_t(request->start_time());
        boost::posix_time::ptime end_time = boost::posix_time::from_time_t(request->over_time());
        // 2. 从数据库中进行数据查询
        auto msg_lists = _mysql_message->range(chat_ssid, start_time, end_time);
        if(msg_lists.empty())
        {
            LOG_ERROR("{} 历史消息为空", rid);
            return error_response(rid, "历史消息为空");
        }
        // 3. 统计所有文件类型 消息的文件ID 并从文件子服务进行批量文件下载
        std::unordered_set<std::string> file_id_lists;
        for(const auto &msg : msg_lists)
        {
            if(msg.file_id().empty()) continue;
            file_id_lists.insert(msg.file_id());
        }
        std::unordered_map<std::string, std::string> file_data_lists;
        bool ret = _GetFile(rid, file_id_lists, file_data_lists);
        if(!ret)
        {
            LOG_ERROR("{} 批量文件下载失败", rid);
            return error_response(rid, "批量文件下载失败");
        }
        // 4. 统计所有消息的发送者用户ID 从用户子服务进行批量下载用户信息
        std::unordered_set<std::string> user_id_lists;
        for(auto &msg : msg_lists)
        {
            user_id_lists.insert(msg.user_id());
        }
        std::unordered_map<std::string, xchat::UserInfo> user_lists;
        ret = _GetUser(rid, user_id_lists, user_lists);
        if(!ret)
        {
            LOG_ERROR("{} 批量用户信息下载失败", rid);
            return error_response(rid, "批量用户信息下载失败");
        }
        // 5.组织响应
        response->set_request_id(rid);
        response->set_success(true);
        for(const auto &msg : msg_lists)
        {
            auto message_info = response->add_msg_list();
            message_info->set_message_id(msg.message_id());
            message_info->set_chat_session_id(msg.session_id());
            message_info->set_timestamp(boost::posix_time::to_time_t(msg.create_time()));
            message_info->mutable_sender()->CopyFrom(user_lists[msg.user_id()]);
            switch(msg.message_type())
            {
                case xchat::MessageType::STRING:
                    message_info->mutable_message()->set_message_type(xchat::MessageType::STRING);
                    message_info->mutable_message()->mutable_string_message()->set_content(msg.content());
                    break;
                case xchat::MessageType::IMAGE:
                    message_info->mutable_message()->set_message_type(xchat::MessageType::IMAGE);
                    message_info->mutable_message()->mutable_image_message()->set_file_id(msg.file_id());
                    message_info->mutable_message()->mutable_image_message()->set_image_content(file_data_lists[msg.file_id()]);
                    break;
                case xchat::MessageType::FILE:
                    message_info->mutable_message()->set_message_type(xchat::MessageType::FILE);
                    message_info->mutable_message()->mutable_file_message()->set_file_id(msg.file_id());
                    message_info->mutable_message()->mutable_file_message()->set_file_name(msg.file_name());
                    message_info->mutable_message()->mutable_file_message()->set_file_size(msg.file_size());
                    message_info->mutable_message()->mutable_file_message()->set_file_contents(file_data_lists[msg.file_id()]);
                    break;
                case xchat::MessageType::SPEECH:
                    message_info->mutable_message()->set_message_type(xchat::MessageType::SPEECH);
                    message_info->mutable_message()->mutable_speech_message()->set_file_id(msg.file_id());
                    message_info->mutable_message()->mutable_speech_message()->set_file_contents(file_data_lists[msg.file_id()]);
                    break;
                default:
                    LOG_ERROR("消息类型错误 {}", msg.message_type());
                    return error_response(rid, "消息类型错误");
            }
        }
        return;

    }
    void GetRecentMsg(::google::protobuf::RpcController* controller,
                       const ::xchat::GetRecentMsgReq* request,
                       ::xchat::GetRecentMsgRsp* response,
                       ::google::protobuf::Closure* done)
    {
        brpc::ClosureGuard rpc_guard(done);
        auto error_response = [this, response](const std::string &rid, const std::string &error_msg) -> void
        {
            response->set_request_id(rid);
            response->set_success(false);
            response->set_errmsg(error_msg);
        };      
        // 1.提取关键要素 请求ID 会话ID 获取消息数量 
        std::string rid = request->request_id();
        std::string chat_ssid = request->chat_session_id();
        int64_t msg_count = request->msg_count();
        // 2.从数据库中获取最近的消息元信息
        auto msg_lists = _mysql_message->recent(chat_ssid, msg_count);
        // 3. 统计所有文件类型 消息的文件ID 并从文件子服务进行批量文件下载
        std::unordered_set<std::string> file_id_lists;
        for(const auto &msg : msg_lists)
        {
            if(msg.file_id().empty()) continue;
            file_id_lists.insert(msg.file_id());
        }
        std::unordered_map<std::string, std::string> file_data_lists;
        bool ret = _GetFile(rid, file_id_lists, file_data_lists);
        if(!ret)
        {
            LOG_ERROR("{} 批量文件下载失败", rid);
            return error_response(rid, "批量文件下载失败");
        }
        // 4. 统计所有消息的发送者用户ID 从用户子服务进行批量下载用户信息
        std::unordered_set<std::string>  user_id_lists;
        for(auto &msg : msg_lists)
        {
            user_id_lists.insert(msg.user_id());
        }
        std::unordered_map<std::string, xchat::UserInfo> user_lists;
        ret = _GetUser(rid, user_id_lists, user_lists);
        if(!ret)
        {
            LOG_ERROR("{} 批量用户信息下载失败", rid);
            return error_response(rid, "批量用户信息下载失败");
        }
        // 5.组织响应
        response->set_request_id(rid);
        response->set_success(true);
        for(const auto &msg : msg_lists)
        {
            auto message_info = response->add_msg_list();
            message_info->set_message_id(msg.message_id());
            message_info->set_chat_session_id(msg.session_id());
            message_info->set_timestamp(boost::posix_time::to_time_t(msg.create_time()));
            message_info->mutable_sender()->CopyFrom(user_lists[msg.user_id()]);
            switch(msg.message_type())
            {
                case xchat::MessageType::STRING:
                    message_info->mutable_message()->set_message_type(xchat::MessageType::STRING);
                    message_info->mutable_message()->mutable_string_message()->set_content(msg.content());
                    break;
                case xchat::MessageType::IMAGE:
                    message_info->mutable_message()->set_message_type(xchat::MessageType::IMAGE);
                    message_info->mutable_message()->mutable_image_message()->set_file_id(msg.file_id());
                    message_info->mutable_message()->mutable_image_message()->set_image_content(file_data_lists[msg.file_id()]);
                    break;
                case xchat::MessageType::FILE:
                    message_info->mutable_message()->set_message_type(xchat::MessageType::FILE);
                    message_info->mutable_message()->mutable_file_message()->set_file_id(msg.file_id());
                    message_info->mutable_message()->mutable_file_message()->set_file_name(msg.file_name());
                    message_info->mutable_message()->mutable_file_message()->set_file_size(msg.file_size());
                    message_info->mutable_message()->mutable_file_message()->set_file_contents(file_data_lists[msg.file_id()]);
                    break;
                case xchat::MessageType::SPEECH:
                    message_info->mutable_message()->set_message_type(xchat::MessageType::SPEECH);
                    message_info->mutable_message()->mutable_speech_message()->set_file_id(msg.file_id());
                    message_info->mutable_message()->mutable_speech_message()->set_file_contents(file_data_lists[msg.file_id()]);
                    break;
                default:
                    LOG_ERROR("消息类型错误 {}", msg.message_type());
                    return error_response(rid, "消息类型错误");
            }
        }
        return;
    }
    void MsgSearch(::google::protobuf::RpcController* controller,
                       const ::xchat::MsgSearchReq* request,
                       ::xchat::MsgSearchRsp* response,
                       ::google::protobuf::Closure* done)
    {
        brpc::ClosureGuard rpc_guard(done);
        auto error_response = [this, response](const std::string &rid, const std::string &error_msg) -> void
        {
            response->set_request_id(rid);
            response->set_success(false);
            response->set_errmsg(error_msg);
        };
        //关键字的消息搜索 只针对文本消息
        // 1.提取关键要素 请求ID 会话ID 关键字 
        std::string rid = request->request_id();
        std::string chat_ssid = request->chat_session_id();
        std::string skey = request->search_key();
        LOG_INFO(" {}->{}->{}", rid, chat_ssid, skey);
        // 2.从ES搜索引擎中进行关键字消息搜索 得到消息列表 --->只有文本消息<----
        auto msg_lists = _es_message->search(skey, chat_ssid);
        // 3. 统计所有消息的发送者用户ID 从用户子服务进行批量下载用户信息
        std::unordered_set<std::string> user_id_lists;
        for(auto &msg : msg_lists)
        {
            user_id_lists.insert(msg.user_id());
        }
        std::unordered_map<std::string, xchat::UserInfo> user_lists;
        bool ret = _GetUser(rid, user_id_lists, user_lists);
        if(!ret)
        {
            LOG_ERROR("{} 批量用户信息下载失败", rid);
            return error_response(rid, "批量用户信息下载失败");
        }
        // 5.组织响应
        response->set_request_id(rid);
        response->set_success(true);
        for(const auto &msg : msg_lists) 
        {
            //--->只有文本消息<----
            auto message_info = response->add_msg_list();
            message_info->set_message_id(msg.message_id());
            message_info->set_chat_session_id(msg.session_id());
            message_info->set_timestamp(boost::posix_time::to_time_t(msg.create_time()));
            message_info->mutable_sender()->CopyFrom(user_lists[msg.user_id()]);
            
            message_info->mutable_message()->set_message_type(xchat::MessageType::STRING);
            message_info->mutable_message()->mutable_string_message()->set_content(msg.content());
        }
        return;
    }
public:
    void onMessage(const char* body, size_t size)
    {
        // 1.取出序列化的消息内容 进行消息处理
        xchat::MessageInfo message;
        bool ret = message.ParseFromArray(body, size);
        if(!ret)
        {
            LOG_ERROR("message 反序列化 失败");
            return;
        }
        // 2.根据不同消息类型进行不同处理
        //   a.文本消息 取元信息到ES中
        //   b.图片/语音/文件消息 存储到文件子服务中 并获取文件ID
        std::string file_content;
        std::string file_id;
        std::string file_name;
        int64_t file_size;
        switch(message.message().message_type())
        {
            //文本消息
            case xchat::MessageType::STRING:
                file_size = message.message().string_message().content().size();
                file_content = message.message().string_message().content();
                ret = _es_message->appendData(
                                    message.sender().user_id(),
                                    message.message_id(),
                                    message.timestamp(),
                                    message.chat_session_id(),
                                    message.message().string_message().content());
                if(!ret)
                {
                    LOG_ERROR("es存储文本消息失败{}", message.message_id());
                    return;
                }
                break;
            //图片消息
            case xchat::MessageType::IMAGE:
            {
                const auto &image = message.message().image_message();
                ret = _PutFile("", image.image_content(), image.image_content().size(), file_id);
                file_size = image.image_content().size();
                //LOG_DEBUG("{} - {}",image.image_content(),  image.image_content().size());
                if (!ret)
                {
                    LOG_ERROR("上传图片到文件子服务失败");
                    return;
                }
                break;
            }
            //文件消息
            case xchat::MessageType::FILE:
            {
                const auto &fmsg = message.message().file_message();
                file_name = fmsg.file_name();
                file_size = fmsg.file_size();
                ret = _PutFile(file_name, fmsg.file_contents(), file_size, file_id);
                if (!ret)
                {
                    LOG_ERROR("上传文件到文件子服务失败");
                    return;
                }
            }
                break;
            //语音消息
            case xchat::MessageType::SPEECH:
            {
                const auto &smsg = message.message().speech_message();
                ret = _PutFile("", smsg.file_contents(), smsg.file_contents().size(), file_id);
                file_size = smsg.file_contents().size();
                if (!ret)
                {
                    LOG_ERROR("上传语音到文件子服务失败");
                    return;
                }
                break;
            }
            default:
                LOG_ERROR("未知消息类型!");
                return;
        }
        // 3.提取出元消息存储到MySQL中
        xchat::Message msg(message.message_id(), message.chat_session_id(),
                            message.sender().user_id(), message.message().message_type(),
                            boost::posix_time::from_time_t(message.timestamp()));
        msg.content(file_content);
        msg.file_id(file_id);
        msg.file_name(file_name);
        msg.file_size(file_size);
        ret = _mysql_message->insert(msg);
        if(!ret)
        {
            LOG_ERROR("mysql存储消息元数据失败");
            return;
        }
    }
private:
    //通过userid 拿到用户id->用户信息内容的映射
    bool _GetUser(const std::string &rid, const std::unordered_set<std::string>  &user_id_lists,
                    std::unordered_map<std::string, xchat::UserInfo> &user_lists)
    {
        auto channel = _mm_channels->choose(_user_service_name);
        if(!channel)
        {
            LOG_ERROR("{}没有可供访问的用户子服务节点", _user_service_name);
            return false;
        }
        UserService_Stub stub(channel.get());
        GetMultiUserInfoReq get_multi_user_req;
        get_multi_user_req.set_request_id(rid);
        for(const auto &user_id : user_id_lists)
        {
            get_multi_user_req.add_users_id(user_id);
        }

        std::unique_ptr<xchat::GetMultiUserInfoRsp> get_multi_user_rsp(new xchat::GetMultiUserInfoRsp());
        std::unique_ptr<brpc::Controller> cntl(new brpc::Controller());
        stub.GetMultiUserInfo(cntl.get(), &get_multi_user_req, get_multi_user_rsp.get(), nullptr); // 同步调用
        if(cntl->Failed())
        {
            LOG_ERROR("获取多个用户失败 错误原因: {}", cntl->ErrorText());
            return false;
        }
        if(!get_multi_user_rsp->success())
        {
            LOG_ERROR("获取多个用户失败 错误原因: {}", get_multi_user_rsp->errmsg());
            return false;
        }
        
        const auto &umap = get_multi_user_rsp->users_info();
        for(auto it = umap.begin(); it != umap.end(); ++it)
        {
            user_lists.insert(std::make_pair(it->first, it->second));
        }
        return true;
    }
    //通过文件id 拿到文件id->文件内容的映射
    bool _GetFile(const std::string &rid, const std::unordered_set<std::string> &file_id_lists,
                    std::unordered_map<std::string, std::string> &file_data_lists)
    {
        auto channel = _mm_channels->choose(_file_service_name);
        if(!channel)
        {
            LOG_ERROR("{}没有可供访问的文件子服务节点", _file_service_name);
            return false;
        }
        FileService_Stub stub(channel.get());
        GetMultiFileReq get_multi_file_req;
        get_multi_file_req.set_request_id(rid);
        for(const auto &file_id : file_id_lists)
        {
            get_multi_file_req.add_file_id_list(file_id);
        }

        std::unique_ptr<xchat::GetMultiFileRsp> get_multi_file_rsp(new xchat::GetMultiFileRsp());
        std::unique_ptr<brpc::Controller> cntl(new brpc::Controller());
        stub.GetMultiFile(cntl.get(), &get_multi_file_req, get_multi_file_rsp.get(), nullptr); // 同步调用
        if(cntl->Failed())
        {
            LOG_ERROR("获取多个文件调用失败 错误原因: {}", cntl->ErrorText());
            return false;
        }
        if(!get_multi_file_rsp->success())
        {
            LOG_ERROR("获取多个文件调用失败 错误原因: {}", get_multi_file_rsp->errmsg());
            return false;
        }
        
        const auto &fmap = get_multi_file_rsp->file_data();
        for(auto it = fmap.begin(); it != fmap.end(); ++it)
        {
            file_data_lists.insert(std::make_pair(it->first, it->second.file_content()));
        }
        return true;

    }
    bool _PutFile(const std::string &file_name, const std::string &body,
                    const int64_t fsize, std::string &file_id)
    {
        //实现文件数据上传
        //LOG_DEBUG("{} - {}",body, fsize);
        auto channel = _mm_channels->choose(_file_service_name);
        if(!channel)
        {
            LOG_ERROR("{}没有可供访问的文件子服务节点", _file_service_name);
            return false;
        }
        FileService_Stub stub(channel.get());
        PutSingleFileReq put_single_file_req;

        put_single_file_req.mutable_file_data()->set_file_name(file_name);
        put_single_file_req.mutable_file_data()->set_file_size(fsize);
        put_single_file_req.mutable_file_data()->set_file_content(body);

        std::unique_ptr<xchat::PutSingleFileRsp> put_single_file_rsp(new xchat::PutSingleFileRsp());
        std::unique_ptr<brpc::Controller> cntl(new brpc::Controller());
        stub.PutSingleFile(cntl.get(), &put_single_file_req, put_single_file_rsp.get(), nullptr); // 同步调用
        if(cntl->Failed())
        {
            LOG_ERROR("上传单个文件调用失败 错误原因: {}", cntl->ErrorText());
            return false;
        }
        if(!put_single_file_rsp->success())
        {
            LOG_ERROR("上传单个文件调用失败 错误原因: {}", put_single_file_rsp->errmsg());
            return false;
        }

        file_id = put_single_file_rsp->file_info().file_id();
        return true;
    }
private:
    //es客户端相关对象
    ESMessage::ptr _es_message; //es用户表
    //mysql客户端相关对象
    MessageTable::ptr _mysql_message; //用户表
    //rpc调用客户端相关对象
    std::string _file_service_name;         //文件子服务名称 方便信道通信
    std::string _user_service_name;         //文件子服务名称 方便信道通信
    ServiceManager::ptr _mm_channels; //信道管理对象
        //消息队列客户端 消息句柄
    // std::string _exchange_name;
    // std::string _binding_key;

};
class MessageServer
{
    public:
        using ptr = std::shared_ptr<MessageServer>;
        MessageServer(const MQClient::ptr& mq_client,
                const Discovery::ptr &service_discovery, 
                const Registry::ptr &reg_client, 
                const std::shared_ptr<elasticlient::Client> &es_client, 
                const std::shared_ptr<odb::core::database> &mysql_client,
                const std::shared_ptr<brpc::Server> &server)
            :_mq_client(mq_client)
            ,_es_client(es_client)
            ,_mysql_client(mysql_client)
            ,_service_discovery(service_discovery)
            ,_reg_client(reg_client)
            ,_rpc_server(server)
        {}
        ~MessageServer()
        {}

        //搭建rpc服务器 并启动服务器
        void start()
        {
            _rpc_server->RunUntilAskedToQuit();
        }
    private:
        MQClient::ptr _mq_client;       //消息队列
        std::shared_ptr<elasticlient::Client> _es_client; //es客户端
        std::shared_ptr<odb::core::database> _mysql_client; //数据库连接
        Discovery::ptr _service_discovery; //服务发现对象
        Registry::ptr _reg_client;  //注册中心
        std::shared_ptr<brpc::Server> _rpc_server;       //rpc服务
};
//建筑师模式
class MessageServerBuilder
{
    public:
        //构造es客户端对象
        void make_es_object(const std::vector<std::string>& host_list)
        {
            _es_client = ESClientFactory::createClient(host_list);
        }
        //构造mysql客户端对象
        void make_mysql_object(const std::string &user, const std::string &pwd, 
                        const std::string &host, const std::string &db,const std::string &cset,
                        int port, int conn_pool_count)
        {
            _mysql_client = ODBFactory::create(user, pwd, host, db, cset, port, conn_pool_count);
        }
        //用于构造服务发现客户端对象 & 信道管理对象
        void make_discovery_object(const std::string &reg_host, const std::string &base_service_name
                                    , const std::string &file_service_name, const std::string &user_service_name)
        {
            _file_service_name = file_service_name;
            _user_service_name = user_service_name;
            _mm_channels = std::make_shared<ServiceManager>();
            _mm_channels->declared(file_service_name);   //声明关心那些服务
            _mm_channels->declared(user_service_name);   //声明关心那些服务
            auto put_cb = std::bind(&ServiceManager::onServiceOnline, _mm_channels.get(), std::placeholders::_1, std::placeholders::_2);
            auto del_cb = std::bind(&ServiceManager::onServiceOffline, _mm_channels.get(), std::placeholders::_1, std::placeholders::_2);
            _service_discovery = std::make_shared<Discovery>(reg_host, base_service_name, put_cb, del_cb);
        }
        //用于构造消息队列客户端对象
        void make_mq_object(const std::string &user, const std::string &pwd, const std::string &host
                        , const std::string &exchange_name, const std::string &queue_name, const std::string &binding_key)
        {
            _exchange_name = exchange_name;
            _binding_key = binding_key;
            _queue_name = queue_name;
            _mq_client = std::make_shared<MQClient>(user, pwd, host);
            _mq_client->declareComponents(exchange_name, queue_name, binding_key);

        }
        //用于构造服务注册客户端对象  服务中心地址 服务名称 当前实例的外部访问地址
        void make_registry_object(const std::string &reg_host, const std::string server_name, const std::string &access_host)
        {
            _reg_client = std::make_shared<Registry>(reg_host);
            _reg_client->registry(server_name, access_host);
        }
        void make_rpc_server(uint16_t port, uint32_t timeout, uint8_t num_threads)
        {
            if(!_es_client)
            {
                LOG_ERROR(" es客户端对象未构造!");
                abort();
            }
            if (!_mysql_client)
            {
                LOG_ERROR("mysql客户端对象未构造");
                abort();
            }
            if(!_mq_client)
            {
                LOG_ERROR("rabbitmq客户端对象未构造");
                abort();
            }
            if(!_mm_channels)
            {
                LOG_ERROR("信道管理对象未初始化");
                abort();
            }
            _rpc_server = std::make_shared<brpc::Server>();
            MessageServiceImpl *msg_service = new MessageServiceImpl(_es_client, _mysql_client
                , _mm_channels, _file_service_name, _user_service_name);
            int ret = _rpc_server->AddService(msg_service, brpc::ServiceOwnership::SERVER_OWNS_SERVICE);
            if(ret == -1)
            {
                LOG_ERROR("添加rpc服务失败!");
                abort();
            }
            brpc::ServerOptions options;
            options.idle_timeout_sec = timeout;  //连接空闲超时时间-超时后连接被关闭
            options.num_threads = num_threads;        //io线程数量
            //4.启动服务器
            ret = _rpc_server->Start(port, &options);
            if(ret == -1)
            {
                LOG_ERROR("启动rpc服务器失败!");
                abort();
            }
            auto callback = std::bind(&MessageServiceImpl::onMessage, msg_service, std::placeholders::_1, std::placeholders::_2);
            //订阅队列 using MessageCallback = std::function<void(const char*, size_t)>;
            _mq_client->consume(_queue_name, callback);
                    
       
        }
        MessageServer::ptr build()
        {
            //_rpc_server 构造前 其余的得构造好

            if(!_service_discovery)
            {
                LOG_ERROR("服务发现对象未初始化");
                abort();
            }
            if(!_reg_client)
            {
                LOG_ERROR("服务发现对象未初始化");
                abort();
            }
            if(!_rpc_server)
            {
                LOG_ERROR("rpc服务对象未初始化");
                abort();
            }
            
            MessageServer::ptr server = std::make_shared<MessageServer>(_mq_client, _service_discovery, 
                _reg_client, _es_client, 
                _mysql_client, _rpc_server);

            
            return server;
        }
    private:
        std::string _file_service_name;                         //文件子服务名称 方便信道通信
        std::string _user_service_name;                         //文件子服务名称 方便信道通信
        Discovery::ptr _service_discovery;                      //服务发现对象
        Registry::ptr _reg_client;                              //注册中心
        std::shared_ptr<brpc::Server> _rpc_server;              //rpc服务
        std::shared_ptr<elasticlient::Client> _es_client;       //es客户端
        std::shared_ptr<odb::core::database> _mysql_client;    //数据库连接
        MQClient::ptr _mq_client;                               //消息队列客户端
        std::string _exchange_name;
        std::string _queue_name;                                //便于订阅队列consume
        std::string _binding_key;

        ServiceManager::ptr _mm_channels;                       //信道管理对象
};
}